package io.kiki.sba.registry.server.data_store.timer;

import io.kiki.sba.registry.common.model.DataUtils;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.LoopExecuteTask;
import io.kiki.sba.registry.util.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class CacheCountTask {
    private static final Logger logger = LoggerFactory.getLogger(CacheCountTask.class);


    @Autowired
    private DatumStorageDelegate datumStorageDelegate;

    @Autowired
    private ServerConfig serverConfig;

    /**
     * instanceId/group - > {info.count,dataInfoId.count}
     */
    private static void printGroupAppCount(String local, String dataCenter, Map<String, Map<String, Map<String, Tuple<Integer, Integer>>>> counts) {
        for (Entry<String, Map<String, Map<String, Tuple<Integer, Integer>>>> count : counts.entrySet()) {
            final String instanceId = count.getKey();
            for (Entry<String, Map<String, Tuple<Integer, Integer>>> groupCounts : count.getValue().entrySet()) {
                final String group = groupCounts.getKey();
                for (Entry<String, Tuple<Integer, Integer>> apps : groupCounts.getValue().entrySet()) {
                    final String app = apps.getKey();
                    Tuple<Integer, Integer> tupleCount = apps.getValue();
                    logger.info("[Pub]local={},{},{},{},{},{},{}", local, dataCenter, instanceId, group, app, tupleCount.o1, tupleCount.o2);
                }
                ConcurrentUtils.sleepUninterruptibly(2, TimeUnit.MILLISECONDS);
            }
        }
    }

    /**
     * instanceId/group - > {info.count,dataInfoId.count}
     */
    private static void printGroupCount(String local, String dataCenter, Map<String, Map<String, Tuple<Integer, Integer>>> counts) {
        for (Map.Entry<String, Map<String, Tuple<Integer, Integer>>> count : counts.entrySet()) {
            final String instanceId = count.getKey();
            Map<String, Tuple<Integer, Integer>> groupCounts = count.getValue();
            for (Entry<String, Tuple<Integer, Integer>> groupCount : groupCounts.entrySet()) {
                final String group = groupCount.getKey();
                Tuple<Integer, Integer> tupleCount = groupCount.getValue();
                Metrics.PUB_GAUGE.labels(local, dataCenter, instanceId, group).set(tupleCount.o1);
                Metrics.PUB_DATA_ID_GAUGE.labels(local, dataCenter, instanceId, group).set(tupleCount.o2);
                logger.info("[PubGroup]local={},{},{},{},{},{}", local, dataCenter, instanceId, group, tupleCount.o1, tupleCount.o2);
            }
        }
    }

    @PostConstruct
    public boolean init() {
        final int intervalSec = serverConfig.getCacheCountIntervalSecs();
        if (intervalSec <= 0) {
            logger.info("cache count off with intervalSecs={}", intervalSec);
            return false;
        }
        ScheduledExecutorService scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("CacheCountTask"));
        scheduledThreadPoolExecutor.scheduleAtFixedRate(this::count, intervalSec, intervalSec, TimeUnit.SECONDS);
        ConcurrentUtils.createDaemonThread("cache-printer", new PrintTotal()).start();
        return true;
    }

    void printTotal() {
        Map<String, Map<String, List<Publisher>>> allMap = datumStorageDelegate.getAllPublisher();

        for (Entry<String, Map<String, List<Publisher>>> entry : allMap.entrySet()) {
            String dataCenter = entry.getKey();
            Map<String, List<Publisher>> map = entry.getValue();
            if (map.isEmpty()) {
                logger.info("[Total]{},pubs={},dataIds={}", dataCenter, 0, 0);
                continue;
            }
            int pubCount = map.values().stream().mapToInt(p -> p.size()).sum();
            logger.info("[Total]{},pubs={},dataIds={}", dataCenter, pubCount, map.size());
        }
    }

    boolean count() {
        try {
            Map<String, Map<String, List<Publisher>>> allMap = datumStorageDelegate.getAllPublisher();
            if (!allMap.isEmpty()) {
                Metrics.PUB_GAUGE.clear();
                Metrics.PUB_DATA_ID_GAUGE.clear();
                for (Entry<String, Map<String, List<Publisher>>> entry : allMap.entrySet()) {
                    final String dataCenter = entry.getKey();
                    List<Publisher> pubs = new ArrayList<>(512);
                    for (List<Publisher> publishers : entry.getValue().values()) {
                        pubs.addAll(publishers);
                    }

                    /** instanceId/group - > {info.count,dataInfoId.count} */
                    Map<String, Map<String, Tuple<Integer, Integer>>> groupCounts = DataUtils.countGroupByInstanceIdGroup(pubs);
                    printGroupCount(serverConfig.getDataCenter(), dataCenter, groupCounts);

                    Map<String, Map<String, Map<String, Tuple<Integer, Integer>>>> counts = DataUtils.countGroupByInstanceIdGroupApp(pubs);
                    printGroupAppCount(serverConfig.getDataCenter(), dataCenter, counts);
                }
            } else {
                logger.info("datum cache is empty");
            }
            return true;
        } catch (Throwable t) {
            logger.error("cache count error", t);
        }
        return false;
    }


    final class PrintTotal extends LoopExecuteTask {

        @Override
        public void _execute_() {
            printTotal();
        }

        @Override
        public void _wait_() {
            // support minute-level statistics
            ConcurrentUtils.sleepUninterruptibly(60, TimeUnit.SECONDS);
        }
    }
}
